use std::sync::Arc;
use std::time::Duration;

use anyhow::Result;
use anyhow::anyhow;
use chrono::Local;
use fudata::db;
use fudata::db::todo;

use fudata::model::kline::KlineList;
use fudata::model::todo::Todo;
use fudata::model::todo::TodoParameter;
use fudata::model::todo::TodoStatus;
use fudata::model::todo::TodoType;
use log::error;
use log::info;
use log::trace;
use tokio::time::Instant;
use zenoh::query::QueryTarget;
use zenoh::selector::Selector;
use zenoh::prelude::r#async::AsyncResolve;

use crate::helper;
use crate::todo::MAX_ALL_SECURITY;

///
/// every day we update the kzz list after 15:30
///
pub async fn generate_todos_sync_klines() -> Result<()> {
    let start = Local::now();
    info!("generate_todos_sync_klines start...");

    let rs = db::security::select(&None, None, None, None, 0, MAX_ALL_SECURITY).await;
    println!("rs={:?}", rs);
    match rs {
        Ok(securities) => {
            let mut i = 0;
            for security in securities {
                let security_id = &security.id_raw();

                let todo_id = Todo::id_from(
                    &[security_id, TodoType::SyncKlines.to_string().as_str()]
                );

                let paras = get_todo_sync_klines_parameters(security_id).await?;
                println!("(generate_todos_sync_klines) parameters={:?}", paras);

                let todo_rs = todo::find(todo_id.as_str()).await;
                match &todo_rs {
                    Ok(v) =>
                        match v {
                            Some(todo) => {
                                // if there is pending todo, then let it be or else update the plan_at to next 1 midnight
                                if todo.is_done() || todo.is_failed() {
                                    let plan_at = helper::now_plus_date_time(0, 0, 0, 0);
                                    let todo_update = Todo {
                                        plan_at: Some(plan_at),
                                        status: Some(TodoStatus::Pending),
                                        parameters: Some(paras),
                                        ..Default::default()
                                    };
                                    let _ = todo::upsert(todo_id.as_str(), &todo_update).await?;
                                }
                            }
                            None => {
                                // create a new todo, today midnight
                                let plan_at = helper::now_plus_date_time(0, 0, 0, 0);
                                let todo_new = Todo {
                                    name: Some(format!("同步证券{}的K线数据", security_id)),
                                    type_: Some(TodoType::SyncKlines),
                                    // updatable fields
                                    plan_at: Some(plan_at),
                                    status: Some(TodoStatus::Pending),
                                    parameters: Some(paras),
                                    created_at: Some(Local::now()),
                                    ..Default::default()
                                };
                                let _ = todo::upsert(todo_id.as_str(), &todo_new).await?;
                            }
                        }
                    Err(e) => error!("{}", e),
                }
            }
        }
        Err(e) => error!("{}", e),
    }
    info!(
        "generate_todos_sync_klines end/ spent seconds={}",
        Local::now().signed_duration_since(start).num_seconds()
    );

    Ok(())
}

async fn get_todo_sync_klines_parameters(security_id: &str) -> Result<Vec<TodoParameter>> {
    let last_updated = db::kline::find_max_last_updated(security_id).await?;
    let pp = vec![
        TodoParameter {
            name: "security_id".to_string(),
            value: security_id.to_string(),
        },
        TodoParameter {
            name: "last_updated".to_string(),
            value: last_updated,
        }
    ];

    Ok(pp)
}

pub async fn sync_klines(session: Arc<zenoh::Session>, worker_no: u8, todo: &Todo) {
    let security_id = db::todo::get_security_id_by_parameter(todo);
    let last_updated = db::todo::get_last_updated_by_parameter(todo);

    let rs = match (security_id, last_updated) {
        (Ok(sid), Ok(lst)) => sync_klines2(session, worker_no, &sid, &lst).await,
        (Ok(_), Err(e2)) => Err(e2),
        (Err(e1), Ok(_)) => Err(e1),
        (Err(e1), Err(e2)) => Err(e1.context(e2)),
    };
    let _ = super::update_todo_with_result(worker_no, rs, todo).await;
}

async fn sync_klines2(
    session: Arc<zenoh::Session>,
    worker_no: u8,
    security_id: &str,
    last_updated: &str
) -> Result<()> {
    trace!(
        "(sync_klines2) Worker[{}] security_id= {}, last_updated= {}",
        worker_no,
        security_id,
        last_updated
    );
    let mut rs: Result<()> = Ok(());
    let selector = Selector::try_from(
        format!("klines?security_id={}&last_updated={}", security_id, last_updated)
    ).unwrap();

    let value = Some(session.zid().to_string());
    let target = QueryTarget::All;
    let timeout = Duration::from_secs(10);

    trace!("\n\n(sync_klines2) Sending Query '{selector}'...\n\n");
    let replies_rs = (
        match value {
            Some(value) => session.get(&selector).with_value(value),
            None => session.get(&selector),
        }
    )
        .target(target)
        .timeout(timeout)
        .res().await;

    match replies_rs {
        Ok(replies) => {
            let start = Instant::now();
            while let Ok(reply) = replies.recv_async().await {
                match reply.sample {
                    Ok(sample) => {
                        trace!(
                            "\n\n (sync_klines2) Worker[{worker_no}] Received ('{}': '{}')",
                            sample.key_expr.as_str(),
                            sample.value
                        );
                        let ss = KlineList::try_from(&sample.value);
                        match ss {
                            Ok(sl) => {
                                for (index, kline) in sl.0.iter().enumerate() {
                                    trace!(
                                        "(sync_klines2) Worker[{worker_no}] Kline[{}], : {:?}",
                                        index,
                                        kline
                                    );
                                    let _rs = db::kline::upsert(security_id, kline).await;
                                }
                            }
                            Err(e) => {
                                error!(
                                    "(sync_klines2) Worker[{worker_no}] KlineList::try_from(sample.value){:?}",
                                    e
                                );
                                rs = Err(anyhow!(e));
                                break;
                            }
                        }
                    }
                    Err(e) => {
                        error!("\n\n (sync_klines2) Worker[{worker_no}] Received (ERROR: '{}')", e);
                        rs = Err(anyhow!("(sync_klines2) Received Error: {:?}", e));
                        break;
                    }
                }
            }
            info!(
                "\n\n (sync_klines2) Worker[{worker_no}] The query has spent {} seconds",
                start.elapsed().as_secs()
            );
        }
        Err(e) => {
            rs = Err(anyhow!("(sync_klines2) {}", e));
        }
    }

    rs
}

#[cfg(test)]
mod test {
    use fudata::db;
    use fudata::model::todo::TodoType;
    use fudata::model::todo::Todo;

    use crate::todo::sync_klines_todo;

    #[tokio::test]
    async fn test_generate_todos_sync_klines() {
        pretty_env_logger::init_timed();
        let _ = fudata::db::init2("127.0.0.1", 8000).await.unwrap();

        let _ = sync_klines_todo::generate_todos_sync_klines().await;

        let todo_id = Todo::id_from(&["SH.KZZ.110047", TodoType::SyncKlines.to_string().as_str()]);

        let rs = db::todo::find(todo_id.as_str()).await;
        println!("rs: {:?}", rs);
    }
}
